package kproduct

import (
	"context"
	"github.com/Shopify/sarama"
	"github.com/rcrowley/go-metrics"
	"log"
	"strings"
	"time"
)

type SyncKafkaProduct struct {
	Brokers 	string

	producer 	sarama.SyncProducer
	cancel 		context.CancelFunc
	logFunc 	func(v ...interface{})
}

func (k *SyncKafkaProduct) Close()  {
	k.cancel()
	return
}

func NewSyncKafkaProduct(brokers string, cancel context.CancelFunc, useNilMetrics bool, logFunc func(v ...interface{})) *SyncKafkaProduct {
	metrics.UseNilMetrics = useNilMetrics
	return &SyncKafkaProduct{
		Brokers: brokers,
		cancel: cancel,
		logFunc: logFunc,
	}
}

func (k *SyncKafkaProduct) SetLogger(logger *log.Logger)  {
	sarama.Logger = logger
	return
}

func (k *SyncKafkaProduct) Start(ctx context.Context) error {
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	config.Producer.Timeout = 5 * time.Second

	p, err := sarama.NewSyncProducer(strings.Split(k.Brokers, ","), config)
	if err != nil {
		return err
	}
	k.producer = p

	k.errLog(ctx)
	return nil
}

func (k *SyncKafkaProduct) errLog(ctx context.Context) {
	go func(p sarama.SyncProducer) {
		for {
			select {
			case <- ctx.Done():
				if k.logFunc != nil {
					k.logFunc("SyncKafkaProduct: Done stop kafka producer, close producer resource")
				}
				p.Close()
				return
			}
		}
	}(k.producer)

	return
}

func (k *SyncKafkaProduct) SendMessage(msg *sarama.ProducerMessage) (partition int32, offset int64, err error) {
	return k.producer.SendMessage(msg)
}

func (k *SyncKafkaProduct) SendMessages(msgs []*sarama.ProducerMessage) error {
	return k.producer.SendMessages(msgs)
}